package com.atguigu.edu.realtime.util;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.bean.TableProcess;
import com.atguigu.edu.realtime.common.Constant;
import com.atguigu.edu.realtime.sink.PhoenixSink;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisSink;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.util.Properties;

public class FlinkSinkUtil {
    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getPhoenixSink() {
        return new PhoenixSink();
    }

    public static SinkFunction<String> getDorisSink(String tableName) {
        Properties props = new Properties();
        props.setProperty("format", "json");
        props.setProperty("strip_outer_array", "true");
        return DorisSink
                .sink(
                        new DorisExecutionOptions.Builder()
                                .setBatchIntervalMs(2000L)
                                .setBatchSize(1024 * 1024)
                                .setEnableDelete(false)
                                .setMaxRetries(3)
                                .setStreamLoadProp(props)
                                .build(),
                        new DorisOptions.Builder()
                                .setFenodes(Constant.DORIS_HOST)
                                .setUsername(Constant.DORIS_USER)
                                .setPassword(Constant.DORIS_PASSWORD)
                                .setTableIdentifier(tableName)
                                .build()
                );
    }
}
